package connector

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"os"
	"sync"
	"time"
)

// FileSourceConnector represents a file source connector
type FileSourceConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// file is the file handle
	file *os.File
	// scanner is the file scanner
	scanner *bufio.Scanner
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// filePath is the file path
	filePath string
	// format is the file format
	format string
}

// NewFileSourceConnector creates a new file source connector
func NewFileSourceConnector() SourceConnector {
	return &FileSourceConnector{
		metrics: make(map[string]interface{}),
		format:  "json",
	}
}

// Initialize initializes the connector
func (c *FileSourceConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get file path
	if path, ok := config["path"].(string); ok {
		c.filePath = path
	} else {
		return fmt.Errorf("path is required")
	}

	// Get format
	if format, ok := config["format"].(string); ok {
		c.format = format
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["records_read"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *FileSourceConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Open the file
	file, err := os.Open(c.filePath)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	c.file = file

	// Create a scanner
	c.scanner = bufio.NewScanner(file)

	// Count lines
	lineCount := 0
	for c.scanner.Scan() {
		lineCount++
	}
	if err := c.scanner.Err(); err != nil {
		return fmt.Errorf("failed to scan file: %w", err)
	}

	// Reset the file
	_, err = file.Seek(0, 0)
	if err != nil {
		return fmt.Errorf("failed to reset file: %w", err)
	}
	c.scanner = bufio.NewScanner(file)

	// Update metrics
	c.metrics["records_total"] = lineCount

	c.started = true

	return nil
}

// Stop stops the connector
func (c *FileSourceConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Close the file
	if c.file != nil {
		err := c.file.Close()
		if err != nil {
			return fmt.Errorf("failed to close file: %w", err)
		}
		c.file = nil
		c.scanner = nil
	}

	c.started = false

	return nil
}

// Read reads records from the source
func (c *FileSourceConnector) Read(ctx context.Context) (*RecordBatch, error) {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return nil, fmt.Errorf("connector not started")
	}

	// Check if scanner is valid
	if c.scanner == nil {
		return nil, fmt.Errorf("scanner not initialized")
	}

	// Get batch size
	batchSize := 10
	if size, ok := c.config["batch_size"].(int); ok {
		batchSize = size
	}

	// Read records
	records := make([]*Record, 0, batchSize)
	for i := 0; i < batchSize && c.scanner.Scan(); i++ {
		line := c.scanner.Text()
		record, err := c.parseRecord(line)
		if err != nil {
			return nil, fmt.Errorf("failed to parse record: %w", err)
		}
		records = append(records, record)
	}

	// Check for scanner error
	if err := c.scanner.Err(); err != nil {
		return nil, fmt.Errorf("failed to scan file: %w", err)
	}

	// Check if no records were read
	if len(records) == 0 {
		return nil, nil
	}

	// Update metrics
	c.metrics["records_read"] = c.metrics["records_read"].(int) + len(records)
	c.metrics["last_read_time"] = time.Now().Unix()

	return NewRecordBatch(records), nil
}

// Commit commits the offset
func (c *FileSourceConnector) Commit(ctx context.Context, offset interface{}) error {
	// File source does not support commit
	return nil
}

// GetMetrics gets connector metrics
func (c *FileSourceConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// parseRecord parses a record from a line
func (c *FileSourceConnector) parseRecord(line string) (*Record, error) {
	// Parse based on format
	switch c.format {
	case "json":
		return c.parseJSONRecord(line)
	case "text":
		return c.parseTextRecord(line)
	default:
		return c.parseJSONRecord(line)
	}
}

// parseJSONRecord parses a JSON record
func (c *FileSourceConnector) parseJSONRecord(line string) (*Record, error) {
	// Parse JSON
	var data map[string]interface{}
	err := json.Unmarshal([]byte(line), &data)
	if err != nil {
		return nil, fmt.Errorf("failed to parse JSON: %w", err)
	}

	// Create record
	record := NewRecord(nil, []byte(line))

	// Set key if provided
	if key, ok := data["key"]; ok {
		if keyStr, ok := key.(string); ok {
			record.Key = []byte(keyStr)
		}
	} else if key, ok := data["id"]; ok {
		if keyStr, ok := key.(string); ok {
			record.Key = []byte(keyStr)
		}
	}

	return record, nil
}

// parseTextRecord parses a text record
func (c *FileSourceConnector) parseTextRecord(line string) (*Record, error) {
	// Create record
	record := NewRecord(nil, []byte(line))
	return record, nil
}

// FileSinkConnector represents a file sink connector
type FileSinkConnector struct {
	// config is the connector configuration
	config map[string]interface{}
	// file is the file handle
	file *os.File
	// writer is the file writer
	writer *bufio.Writer
	// metrics are the connector metrics
	metrics map[string]interface{}
	// mu is the mutex
	mu sync.RWMutex
	// started indicates if the connector is started
	started bool
	// filePath is the file path
	filePath string
	// format is the file format
	format string
	// append indicates if the file should be appended
	append bool
}

// NewFileSinkConnector creates a new file sink connector
func NewFileSinkConnector() SinkConnector {
	return &FileSinkConnector{
		metrics: make(map[string]interface{}),
		format:  "json",
		append:  false,
	}
}

// Initialize initializes the connector
func (c *FileSinkConnector) Initialize(config map[string]interface{}) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Store the configuration
	c.config = config

	// Get file path
	if path, ok := config["path"].(string); ok {
		c.filePath = path
	} else {
		return fmt.Errorf("path is required")
	}

	// Get format
	if format, ok := config["format"].(string); ok {
		c.format = format
	}

	// Get append
	if append, ok := config["append"].(bool); ok {
		c.append = append
	}

	// Initialize metrics
	c.metrics["records_total"] = 0
	c.metrics["start_time"] = time.Now().Unix()

	return nil
}

// Start starts the connector
func (c *FileSinkConnector) Start(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already started
	if c.started {
		return fmt.Errorf("connector already started")
	}

	// Open the file
	flag := os.O_CREATE | os.O_WRONLY
	if c.append {
		flag |= os.O_APPEND
	} else {
		flag |= os.O_TRUNC
	}
	file, err := os.OpenFile(c.filePath, flag, 0644)
	if err != nil {
		return fmt.Errorf("failed to open file: %w", err)
	}
	c.file = file

	// Create a writer
	c.writer = bufio.NewWriter(file)

	c.started = true

	return nil
}

// Stop stops the connector
func (c *FileSinkConnector) Stop(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if already stopped
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Flush the writer
	if c.writer != nil {
		err := c.writer.Flush()
		if err != nil {
			return fmt.Errorf("failed to flush writer: %w", err)
		}
	}

	// Close the file
	if c.file != nil {
		err := c.file.Close()
		if err != nil {
			return fmt.Errorf("failed to close file: %w", err)
		}
		c.file = nil
		c.writer = nil
	}

	c.started = false

	return nil
}

// Write writes records to the sink
func (c *FileSinkConnector) Write(ctx context.Context, batch *RecordBatch) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if batch is valid
	if batch == nil || len(batch.Records) == 0 {
		return nil
	}

	// Check if writer is valid
	if c.writer == nil {
		return fmt.Errorf("writer not initialized")
	}

	// Write records
	for _, record := range batch.Records {
		// Format record
		var output string
		var err error
		switch c.format {
		case "json":
			output, err = c.formatJSON(record)
		case "text":
			output, err = c.formatText(record)
		default:
			output, err = c.formatJSON(record)
		}

		if err != nil {
			return fmt.Errorf("failed to format record: %w", err)
		}

		// Write record
		_, err = c.writer.WriteString(output + "\n")
		if err != nil {
			return fmt.Errorf("failed to write record: %w", err)
		}
	}

	// Update metrics
	c.metrics["records_total"] = c.metrics["records_total"].(int) + len(batch.Records)
	c.metrics["last_write_time"] = time.Now().Unix()

	return nil
}

// Flush flushes any buffered records
func (c *FileSinkConnector) Flush(ctx context.Context) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Check if started
	if !c.started {
		return fmt.Errorf("connector not started")
	}

	// Check if writer is valid
	if c.writer == nil {
		return fmt.Errorf("writer not initialized")
	}

	// Flush the writer
	err := c.writer.Flush()
	if err != nil {
		return fmt.Errorf("failed to flush writer: %w", err)
	}

	return nil
}

// GetMetrics gets connector metrics
func (c *FileSinkConnector) GetMetrics() map[string]interface{} {
	c.mu.RLock()
	defer c.mu.RUnlock()

	// Copy metrics
	metrics := make(map[string]interface{})
	for k, v := range c.metrics {
		metrics[k] = v
	}

	return metrics
}

// formatJSON formats a record as JSON
func (c *FileSinkConnector) formatJSON(record *Record) (string, error) {
	// Parse value as JSON
	var value interface{}
	err := json.Unmarshal(record.Value, &value)
	if err != nil {
		// If not valid JSON, use as string
		value = string(record.Value)
	}

	// Create output
	output := map[string]interface{}{
		"key":       string(record.Key),
		"value":     value,
		"timestamp": record.Timestamp,
	}

	// Add headers if not empty
	if len(record.Headers) > 0 {
		output["headers"] = record.Headers
	}

	// Add metadata if not empty
	if len(record.Metadata) > 0 {
		output["metadata"] = record.Metadata
	}

	// Marshal to JSON
	data, err := json.Marshal(output)
	if err != nil {
		return "", fmt.Errorf("failed to marshal record: %w", err)
	}

	return string(data), nil
}

// formatText formats a record as text
func (c *FileSinkConnector) formatText(record *Record) (string, error) {
	// Format key
	key := string(record.Key)
	if key == "" {
		key = "<null>"
	}

	// Format value
	value := string(record.Value)
	if value == "" {
		value = "<null>"
	}

	// Format timestamp
	timestamp := record.Timestamp.Format(time.RFC3339)

	// Create output
	output := fmt.Sprintf("%s,%s,%s", key, value, timestamp)

	return output, nil
}
